package com.yungui.kafka.consumer.config;

import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.client.DefaultConnectionKeepAliveStrategy;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.List;

@Configuration
public class ESConfig {

	@Value("${elasticsearch.nodes}")
	private List<String> nodes;

	@Value("${elasticsearch.schema}")
	private String schema;

	@Value("${elasticsearch.max-connect-total}")
	private Integer maxConnectTotal;

	@Value("${elasticsearch.max-connect-per-route}")
	private Integer maxConnectPerRoute;

	@Value("${elasticsearch.connection-request-timeout-millis}")
	private Integer connectionRequestTimeoutMillis;

	@Value("${elasticsearch.socket-timeout-millis}")
	private Integer socketTimeoutMillis;

	@Value("${elasticsearch.connect-timeout-millis}")
	private Integer connectTimeoutMillis;

	@Value("${elasticsearch.username}")
	private String username;

	@Value("${elasticsearch.password}")
	private String password;

	@Value("${elasticsearch.max-retry-timeout-millis}")
	private Integer maxRetryTimeoutMillis;

	@Bean
	public RestHighLevelClient getRestHighLevelClient() {

		//1、获取主机地址
		List<HttpHost> httpHosts = new ArrayList<>();
		for (String node : nodes) {
			try {
				String[] parts = StringUtils.split(node, ":");
				Assert.notNull(parts, "Must defined");
				Assert.state(parts.length == 2, "Must be defined as 'host:port'");
				httpHosts.add(new HttpHost(parts[0], Integer.parseInt(parts[1]), schema));
			} catch (RuntimeException ex) {
				throw new IllegalStateException("Invalid ES nodes " + "property '" + node + "'", ex);
			}
		}
		HttpHost[] httpHostArr = httpHosts.toArray(new HttpHost[nodes.size()]);
		//2、配置认证
		final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
		credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));

		//3、创建RestHighLevelClient
		RestClientBuilder builder = RestClient.builder(httpHostArr);
		//设置最大超时重试
		builder.setRequestConfigCallback(requestConfigBuilder -> {
			//与服务器连接超时时间
			requestConfigBuilder.setConnectTimeout(connectTimeoutMillis);
			//socket读数据超时时间：从服务器获取响应数据的超时时间
			requestConfigBuilder.setSocketTimeout(socketTimeoutMillis);
			//从连接池中获取连接的超时时间
			requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeoutMillis);
			return requestConfigBuilder;
		});

		builder.setHttpClientConfigCallback(httpClientBuilder -> {
			httpClientBuilder.setKeepAliveStrategy(new DefaultConnectionKeepAliveStrategy());
			httpClientBuilder.setMaxConnTotal(maxConnectTotal);
			httpClientBuilder.setMaxConnPerRoute(maxConnectPerRoute);
//			httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
			return httpClientBuilder;
		});

		return new RestHighLevelClient(builder);
	}
}
